package com.neu.mapreduce.tweet;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BinaryComparable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/**
 * @author anshuman kumar
 */
public class LocationAnalysis {

	/**
	 * @author anshuman kumar
	 * 
	 */
	public static class LocationAnalysisMapper extends Mapper<Object, Text, Text, IntWritable> {

		public static final int MAX_RECORD = 10;
		private Map<String, Integer> data;

		protected void setup(org.apache.hadoop.mapreduce.Mapper<Object, Text, Text, IntWritable>.Context context)
		        throws IOException, InterruptedException {
			data = new HashMap<String, Integer>();
		}

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] line = value.toString().split("\t");
			if (line.length > 9) {
				String loc = line[8];
				if (loc.trim().length() != 0 && loc.indexOf("&") == -1 && loc.indexOf("-") == -1
				        && loc.indexOf(":") == -1) {
					loc = loc.replace(",", "");
					loc = loc.replace(".", "");
					loc = loc.toLowerCase();
					int count = 0;
					if (data.containsKey(loc)) {
						count = data.get(loc);
					}
					count++;
					data.put(loc, count);
				}
			}
		}

		public void cleanup(org.apache.hadoop.mapreduce.Mapper<Object, Text, Text, IntWritable>.Context context)
		        throws IOException, InterruptedException {

			List<Map.Entry<String, Integer>> as = new ArrayList<Map.Entry<String, Integer>>(data.entrySet());

			Collections.sort(as, new Comparator<Map.Entry<String, Integer>>() {
				@Override
				public int compare(Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) {
					int first = e1.getValue();
					int second = e2.getValue();
					return (second - first);
				}
			});

			int count = 1;
			for (Entry<String, Integer> entry : as) {
				if (count > MAX_RECORD) {
					break;
				}
				KeyComprator word = new KeyComprator();
				// emit({location l:count c},count c)
				word.set(entry.getKey() + ":" + entry.getValue());
				context.write(word, new IntWritable(entry.getValue()));
				count++;
			}

		}

	}

	/**
	 * @author anshuman kumar
	 * 
	 */
	public static class LocationAnalysisReducer extends Reducer<KeyComprator, IntWritable, Text, IntWritable> {

		public void reduce(KeyComprator key, Iterable<IntWritable> values, Context context) throws IOException,
		        InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			KeyComprator first = new KeyComprator();
			first.set(key.toString().split(":")[0]);
			// result.set(Integer.parseInt(key.toString().split(":")[1]));
			context.write(first, new IntWritable(sum));
		}

	}

	/**
	 * @author hduser
	 * 
	 */
	public static class KeyComprator extends Text {

		protected KeyComprator() {
			super();
		}

		@Override
		public int compareTo(BinaryComparable other) {
			int t1Base = Integer.parseInt(this.toString().split(":")[1]);
			int t2Base = Integer.parseInt(((Text) other).toString().split(":")[1]);

			int comp = t2Base - t1Base;
			// descending value
			if (comp == 0) {
				comp = this.toString().split(":")[0].compareTo(((Text) other).toString().split(":")[0]);
			}
			return comp;
		}

	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: com.neu.mapreduce.tweet.LocationAnalysis <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "Location Analysis");
		job.setJarByClass(LocationAnalysis.class);
		job.setMapperClass(LocationAnalysisMapper.class);
		job.setReducerClass(LocationAnalysisReducer.class);
		job.setOutputKeyClass(KeyComprator.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
